package com.kevin.elastic.job;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;

public class ElasticJobDemo {

	public ElasticJobDemo() {
		// TODO Auto-generated constructor stub
	}

	public static void main(String[] args) {
		// 定义作业核心配置
		JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder("SimpleJobDemo", "0/15 * * * * ?", 10).build();
		JobCoreConfiguration simpleCoreConfig2 = JobCoreConfiguration.newBuilder("SimpleJobDemo2", "0/15 * * * * ?", 10).build();
		JobCoreConfiguration simpleCoreConfig3 = JobCoreConfiguration.newBuilder("SimpleJobDemo3", "0/15 * * * * ?", 10).build();
		JobCoreConfiguration simpleCoreConfig4 = JobCoreConfiguration.newBuilder("SimpleJobDemo4", "0/15 * * * * ?", 10).build();
		// 定义SIMPLE类型配置
		SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig,SimpleJobDemo.class.getCanonicalName());
		SimpleJobConfiguration simpleJobConfig2 = new SimpleJobConfiguration(simpleCoreConfig2,SimpleJobDemo2.class.getCanonicalName());
		SimpleJobConfiguration simpleJobConfig3 = new SimpleJobConfiguration(simpleCoreConfig3,SimpleJobDemo3.class.getCanonicalName());
		SimpleJobConfiguration simpleJobConfig4 = new SimpleJobConfiguration(simpleCoreConfig4,SimpleJobDemo4.class.getCanonicalName());
		// 定义Lite作业根配置
		LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
		LiteJobConfiguration simpleJobRootConfig2 = LiteJobConfiguration.newBuilder(simpleJobConfig2).build();
		LiteJobConfiguration simpleJobRootConfig3 = LiteJobConfiguration.newBuilder(simpleJobConfig3).build();
		LiteJobConfiguration simpleJobRootConfig4 = LiteJobConfiguration.newBuilder(simpleJobConfig4).build();

		// 定义作业核心配置
		JobCoreConfiguration dataflowCoreConfig = JobCoreConfiguration.newBuilder("DataflowElasticJob", "0/30 * * * * ?",
				10).build();
		// 定义DATAFLOW类型配置
		DataflowJobConfiguration dataflowJobConfig = new DataflowJobConfiguration(dataflowCoreConfig,
				DataflowElasticJob.class.getCanonicalName(), true);
		// 定义Lite作业根配置
		LiteJobConfiguration dataflowJobRootConfig = LiteJobConfiguration.newBuilder(dataflowJobConfig).build();

		// // 定义作业核心配置配置
		// JobCoreConfiguration scriptCoreConfig = JobCoreConfiguration.newBuilder("demoScriptJob", "0/45 * * * * ?",
		// 10)
		// .build();
		// // 定义SCRIPT类型配置
		// ScriptJobConfiguration scriptJobConfig = new ScriptJobConfiguration(scriptCoreConfig, "test.sh");
		// // 定义Lite作业根配置
		// JobRootConfiguration scriptJobRootConfig = LiteJobConfiguration.newBuilder(scriptCoreConfig).build();


		ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("10.198.195.147:2181", "namespace");
//		zkConfig.setConnectionTimeoutMilliseconds(Integer.MAX_VALUE);
//		zkConfig.setBaseSleepTimeMilliseconds(Integer.MAX_VALUE);
//		zkConfig.setSessionTimeoutMilliseconds(Integer.MAX_VALUE);
//		zkConfig.setMaxSleepTimeMilliseconds(Integer.MAX_VALUE);
//		zkConfig.setMaxRetries(3);
		ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);
		regCenter.init();

		JobElasticJobListener jobListener = new JobElasticJobListener();

		JobScheduler simpleJobScheduler = new JobScheduler(regCenter, simpleJobRootConfig, jobListener);
		simpleJobScheduler.init();
		JobScheduler simpleJobScheduler2 = new JobScheduler(regCenter, simpleJobRootConfig2, jobListener);
		simpleJobScheduler2.init();
		JobScheduler simpleJobScheduler3 = new JobScheduler(regCenter, simpleJobRootConfig3, jobListener);
		simpleJobScheduler3.init();
		JobScheduler simpleJobScheduler4 = new JobScheduler(regCenter, simpleJobRootConfig4, jobListener);
		simpleJobScheduler4.init();
		JobScheduler dataFlowJobScheduler = new JobScheduler(regCenter, dataflowJobRootConfig, jobListener);
		dataFlowJobScheduler.init();

		try {
			Thread.currentThread().wait();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
